#include <gtest/gtest.h>
#include "utils_block_queue.h"
#include "utils_log.h"
#include <thread>
#include <chrono>
#include <vector>
#include <atomic>
#include <string>

using namespace El::Utils;

class BlockingQueueTest : public ::testing::Test {
protected:
    void SetUp() override
    {
        // 每个测试前的设置
    }

    void TearDown() override
    {
        // 每个测试后的清理
    }
};

// 测试基本的Push和Pop操作
TEST_F(BlockingQueueTest, BasicPushPop)
{
    BlockingQueue<int> queue("TestQueue", 5);

    // 测试初始状态
    EXPECT_EQ(queue.Size(), 0);
    EXPECT_EQ(queue.Capacity(), 5);
    EXPECT_TRUE(queue.Empty());

    // 测试Push操作
    EXPECT_TRUE(queue.Push(1));
    EXPECT_TRUE(queue.Push(2));
    EXPECT_TRUE(queue.Push(3));

    EXPECT_EQ(queue.Size(), 3);
    EXPECT_FALSE(queue.Empty());

    // 测试Pop操作
    EXPECT_EQ(queue.Pop(), 1);
    EXPECT_EQ(queue.Pop(), 2);
    EXPECT_EQ(queue.Pop(), 3);

    EXPECT_EQ(queue.Size(), 0);
    EXPECT_TRUE(queue.Empty());
}

// 测试队列容量限制
TEST_F(BlockingQueueTest, CapacityLimit)
{
    BlockingQueue<int> queue("TestQueue", 3);

    // 填满队列
    EXPECT_TRUE(queue.Push(1));
    EXPECT_TRUE(queue.Push(2));
    EXPECT_TRUE(queue.Push(3));
    EXPECT_EQ(queue.Size(), 3);

    // 启动一个线程尝试Push，应该会阻塞
    std::atomic<bool> pushCompleted(false);
    std::thread pushThread([&queue, &pushCompleted]() {
        EXPECT_TRUE(queue.Push(4));
        pushCompleted = true;
    });

    // 等待一小段时间，确保线程已经启动并阻塞
    std::this_thread::sleep_for(std::chrono::milliseconds(100));
    EXPECT_FALSE(pushCompleted);

    // Pop一个元素，应该让Push线程继续执行
    EXPECT_EQ(queue.Pop(), 1);

    // 等待Push线程完成
    pushThread.join();
    EXPECT_TRUE(pushCompleted);
    EXPECT_EQ(queue.Size(), 3);
}

// 测试带超时的Push和Pop操作
TEST_F(BlockingQueueTest, TimeoutOperations)
{
    BlockingQueue<int> queue("TestQueue", 2);

    // 测试Push超时
    EXPECT_TRUE(queue.Push(1));
    EXPECT_TRUE(queue.Push(2));

    // 带超时的Push应该失败
    auto start = std::chrono::steady_clock::now();
    EXPECT_FALSE(queue.Push(3, 100));
    auto duration = std::chrono::duration_cast<std::chrono::milliseconds>(std::chrono::steady_clock::now() - start);
    EXPECT_GE(duration.count(), 90);
    EXPECT_EQ(queue.Size(), 2); // 确保没有插入新元素

    // 测试Pop超时
    queue.Clear();
    start = std::chrono::steady_clock::now();
    auto result = queue.Pop(100);
    duration = std::chrono::duration_cast<std::chrono::milliseconds>(std::chrono::steady_clock::now() - start);
    EXPECT_EQ(result, int());
    EXPECT_GE(duration.count(), 90);
}

// 测试Clear操作
TEST_F(BlockingQueueTest, ClearOperation)
{
    BlockingQueue<int> queue("TestQueue", 5);

    // 添加一些元素
    EXPECT_TRUE(queue.Push(1));
    EXPECT_TRUE(queue.Push(2));
    EXPECT_TRUE(queue.Push(3));
    EXPECT_EQ(queue.Size(), 3);

    // 清空队列
    queue.Clear();
    EXPECT_EQ(queue.Size(), 0);
    EXPECT_TRUE(queue.Empty());

    // 确认可以继续使用队列
    EXPECT_TRUE(queue.Push(4));
    EXPECT_EQ(queue.Size(), 1);
}

// 测试SetActive操作
TEST_F(BlockingQueueTest, SetActiveOperation)
{
    BlockingQueue<int> queue("TestQueue", 5);

    // 添加一些元素
    EXPECT_TRUE(queue.Push(1));
    EXPECT_TRUE(queue.Push(2));

    // 设置为非活动状态
    queue.SetActive(false);

    // 测试Push操作应该失败
    EXPECT_FALSE(queue.Push(3));

    // 测试Pop操作应该返回默认值
    EXPECT_EQ(queue.Pop(), int());

    // 重新激活队列
    queue.SetActive(true);

    // 测试可以继续使用队列
    EXPECT_TRUE(queue.Push(4));
    EXPECT_EQ(queue.Pop(), 4);
}

// 测试多线程并发操作
TEST_F(BlockingQueueTest, MultithreadedOperations)
{
    EL_INFO("开始执行 MultithreadedOperations 测试");
    BlockingQueue<int> queue("TestQueue", 100);
    const int numProducers = 5;
    const int numConsumers = 5;
    const int itemsPerProducer = 100;

    std::atomic<int> producedCount(0);
    std::atomic<int> consumedCount(0);

    EL_INFO("创建 {} 个生产者线程和 {} 个消费者线程，每个生产者产生 {} 个项目", numProducers, numConsumers,
             itemsPerProducer);

    // 启动生产者线程
    std::vector<std::thread> producers;
    for (int i = 0; i < numProducers; ++i) {
        producers.emplace_back([&queue, &producedCount, i, itemsPerProducer]() {
            EL_INFO("生产者线程 {} 开始运行", i);
            for (int j = 0; j < itemsPerProducer; ++j) {
                // 加1确保不会产生值为0的项目
                int value = i * itemsPerProducer + j + 1;

                if (j % 20 == 0) {
                    EL_INFO("生产者 {} 正在生产第 {}/{} 个项目，值为 {}", i, j, itemsPerProducer, value);
                }

                bool success = queue.Push(value);

                if (!success) {
                    EL_ERROR("生产者 {} 无法推送值 {}", i, value);
                } else {
                    int newCount = ++producedCount;
                    if (newCount % 100 == 0) {
                        EL_INFO("已生产 {} 个项目，队列当前大小: {}", newCount, queue.Size());
                    }
                }
            }
            EL_INFO("生产者线程 {} 结束", i);
        });
    }

    // 启动消费者线程
    std::vector<std::thread> consumers;
    for (int i = 0; i < numConsumers; ++i) {
        consumers.emplace_back([&queue, &consumedCount, itemsPerProducer, numProducers, i]() {
            EL_INFO("消费者线程 {} 开始运行", i);
            int localConsumed = 0;

            // 使用原子变量的局部拷贝检查终止条件，避免竞争条件
            while (true) {
                // 先检查是否已经完成，防止多余的pop调用
                int currentConsumed = consumedCount.load();
                if (currentConsumed >= itemsPerProducer * numProducers) {
                    EL_INFO("消费者 {} 检测到消费已完成，跳出循环，当前消费计数: {}", i, currentConsumed);
                    break;
                }

                // 尝试从队列中取出一项
                int value = queue.Pop(10); // 缩短超时时间，更频繁地检查终止条件

                if (value != int()) {
                    localConsumed++;
                    int newCount = ++consumedCount;

                    if (localConsumed % 20 == 0) {
                        EL_INFO("消费者 {} 已消费 {} 个项目，最近消费的值: {}", i, localConsumed, value);
                    }

                    if (newCount % 100 == 0) {
                        EL_INFO("已消费 {} 个项目，队列当前大小: {}", newCount, queue.Size());
                    }

                    // 再次检查是否已经完成消费
                    if (newCount >= itemsPerProducer * numProducers) {
                        EL_INFO("消费者 {} 检测到消费已完成，跳出循环", i);
                        break;
                    }
                } else {
                    EL_INFO("消费者 {} 尝试 Pop 但返回了 0 值", i);
                    std::this_thread::sleep_for(std::chrono::milliseconds(5));
                }

                // 定期打印进度
                if (localConsumed % 50 == 0 && localConsumed > 0) {
                    EL_INFO("消费者 {} 进度: 局部已消费 {}, 全局已消费 {}/{}", i, localConsumed, consumedCount.load(),
                             itemsPerProducer * numProducers);
                }
            }
            EL_INFO("消费者线程 {} 结束，消费了 {} 个项目", i, localConsumed);
        });
    }

    EL_INFO("等待所有线程完成");

    // 等待所有生产者线程完成
    for (size_t i = 0; i < producers.size(); ++i) {
        EL_INFO("等待生产者 {} 完成", i);
        producers[i].join();
        EL_INFO("生产者 {} 已完成", i);
    }

    EL_INFO("所有生产者已完成，当前已生产: {}, 已消费: {}, 队列大小: {}", producedCount.load(), consumedCount.load(),
             queue.Size());

    // 等待所有消费者线程完成
    for (size_t i = 0; i < consumers.size(); ++i) {
        EL_INFO("等待消费者 {} 完成", i);
        if (consumers[i].joinable()) {
            consumers[i].join();
            EL_INFO("消费者 {} 已完成", i);
        } else {
            EL_ERROR("消费者 {} 不可加入", i);
        }
    }

    EL_INFO("测试完成，验证结果");
    EL_INFO("生产项目数: {}, 消费项目数: {}, 队列大小: {}", producedCount.load(), consumedCount.load(), queue.Size());

    // 验证所有项目都被消费
    EXPECT_EQ(producedCount.load(), itemsPerProducer * numProducers);
    EXPECT_EQ(consumedCount.load(), itemsPerProducer * numProducers);
    EXPECT_TRUE(queue.Empty());

    EL_INFO("MultithreadedOperations 测试通过");
}

// 测试大规模多线程并发操作
TEST_F(BlockingQueueTest, LargeScaleMultithreadedOperations)
{
    EL_INFO("开始执行 LargeScaleMultithreadedOperations 测试");
    BlockingQueue<int> queue("LargeTestQueue", 200); // 更大的队列容量
    const int numProducers = 8;     // 更多的生产者
    const int numConsumers = 8;     // 更多的消费者
    const int itemsPerProducer = 5000; // 每个生产者产生更多项目

    std::atomic<int> producedCount(0);
    std::atomic<int> consumedCount(0);

    EL_INFO("创建 {} 个生产者线程和 {} 个消费者线程，每个生产者产生 {} 个项目", 
             numProducers, numConsumers, itemsPerProducer);

    // 启动生产者线程
    std::vector<std::thread> producers;
    for (int i = 0; i < numProducers; ++i) {
        producers.emplace_back([&queue, &producedCount, i, itemsPerProducer]() {
            EL_INFO("生产者线程 {} 开始运行", i);
            for (int j = 0; j < itemsPerProducer; ++j) {
                // 加1确保不会产生值为0的项目
                int value = i * itemsPerProducer + j + 1;

                if (j % 50 == 0) {  // 减少日志数量，每50个项目记录一次
                    EL_INFO("生产者 {} 正在生产第 {}/{} 个项目，值为 {}", 
                             i, j, itemsPerProducer, value);
                }

                bool success = queue.Push(value);

                if (!success) {
                    EL_ERROR("生产者 {} 无法推送值 {}", i, value);
                } else {
                    int newCount = ++producedCount;
                    if (newCount % 500 == 0) {  // 更大的间隔来记录总进度
                        EL_INFO("已生产 {} 个项目，队列当前大小: {}", newCount, queue.Size());
                    }
                }
            }
            EL_INFO("生产者线程 {} 结束", i);
        });
    }

    // 启动消费者线程
    std::vector<std::thread> consumers;
    for (int i = 0; i < numConsumers; ++i) {
        consumers.emplace_back([&queue, &consumedCount, itemsPerProducer, numProducers, i]() {
            EL_INFO("消费者线程 {} 开始运行", i);
            int localConsumed = 0;

            // 使用原子变量的局部拷贝检查终止条件，避免竞争条件
            while (true) {
                // 先检查是否已经完成，防止多余的pop调用
                int currentConsumed = consumedCount.load();
                if (currentConsumed >= itemsPerProducer * numProducers) {
                    EL_INFO("消费者 {} 检测到消费已完成，跳出循环，当前消费计数: {}", 
                             i, currentConsumed);
                    break;
                }

                // 尝试从队列中取出一项
                int value = queue.Pop(10); // 缩短超时时间，更频繁地检查终止条件

                if (value != int()) {
                    localConsumed++;
                    int newCount = ++consumedCount;

                    if (localConsumed % 50 == 0) {  // 减少日志数量，每50个项目记录一次
                        EL_INFO("消费者 {} 已消费 {} 个项目，最近消费的值: {}", 
                                i, localConsumed, value);
                    }

                    if (newCount % 500 == 0) {  // 更大的间隔来记录总进度
                        EL_INFO("已消费 {} 个项目，队列当前大小: {}", newCount, queue.Size());
                    }

                    // 再次检查是否已经完成消费
                    if (newCount >= itemsPerProducer * numProducers) {
                        EL_INFO("消费者 {} 检测到消费已完成，跳出循环", i);
                        break;
                    }
                } else {
                    // 减少日志数量，不记录每次超时
                    if (localConsumed % 100 == 0) {
                        EL_INFO("消费者 {} 尝试 Pop 但返回了 0 值", i);
                    }
                    std::this_thread::sleep_for(std::chrono::milliseconds(5));
                }

                // 定期打印进度
                if (localConsumed % 100 == 0 && localConsumed > 0) {  // 减少日志数量
                    EL_INFO("消费者 {} 进度: 局部已消费 {}, 全局已消费 {}/{}", 
                            i, localConsumed, consumedCount.load(), 
                            itemsPerProducer * numProducers);
                }
            }
            EL_INFO("消费者线程 {} 结束，消费了 {} 个项目", i, localConsumed);
        });
    }

    EL_INFO("等待所有线程完成");

    // 等待所有生产者线程完成
    for (size_t i = 0; i < producers.size(); ++i) {
        EL_INFO("等待生产者 {} 完成", i);
        producers[i].join();
        EL_INFO("生产者 {} 已完成", i);
    }

    EL_INFO("所有生产者已完成，当前已生产: {}, 已消费: {}, 队列大小: {}", 
             producedCount.load(), consumedCount.load(), queue.Size());

    // 等待所有消费者线程完成
    for (size_t i = 0; i < consumers.size(); ++i) {
        EL_INFO("等待消费者 {} 完成", i);
        if (consumers[i].joinable()) {
            consumers[i].join();
            EL_INFO("消费者 {} 已完成", i);
        } else {
            EL_ERROR("消费者 {} 不可加入", i);
        }
    }

    EL_INFO("测试完成，验证结果");
    EL_INFO("生产项目数: {}, 消费项目数: {}, 队列大小: {}", 
             producedCount.load(), consumedCount.load(), queue.Size());

    // 验证所有项目都被消费
    EXPECT_EQ(producedCount.load(), itemsPerProducer * numProducers);
    EXPECT_EQ(consumedCount.load(), itemsPerProducer * numProducers);
    EXPECT_TRUE(queue.Empty());

    EL_INFO("LargeScaleMultithreadedOperations 测试通过");
}

// 测试不同类型的数据
TEST_F(BlockingQueueTest, DifferentDataTypes)
{
    // 测试字符串类型
    {
        BlockingQueue<std::string> queue("StringQueue", 5);
        EXPECT_TRUE(queue.Push("Hello"));
        EXPECT_TRUE(queue.Push("World"));
        EXPECT_EQ(queue.Pop(), "Hello");
        EXPECT_EQ(queue.Pop(), "World");
    }

    // 测试自定义结构体
    {
        struct TestData {
            int id;
            std::string name;

            bool operator==(const TestData &other) const
            {
                return id == other.id && name == other.name;
            }
        };

        BlockingQueue<TestData> queue("StructQueue", 5);
        TestData data1 = {1, "Test1"};
        TestData data2 = {2, "Test2"};

        EXPECT_TRUE(queue.Push(data1));
        EXPECT_TRUE(queue.Push(data2));

        EXPECT_EQ(queue.Pop(), data1);
        EXPECT_EQ(queue.Pop(), data2);
    }
}

// 测试阻塞行为
TEST_F(BlockingQueueTest, BlockingBehavior)
{
    BlockingQueue<int> queue("TestQueue", 2);

    // 测试空队列的Pop阻塞
    std::atomic<bool> popCompleted(false);
    std::thread popThread([&queue, &popCompleted]() {
        queue.Pop();
        popCompleted = true;
    });

    // 等待一小段时间，确保线程已经启动并阻塞
    std::this_thread::sleep_for(std::chrono::milliseconds(100));
    EXPECT_FALSE(popCompleted);

    // Push一个元素，应该让Pop线程继续执行
    EXPECT_TRUE(queue.Push(1));

    // 等待Pop线程完成
    popThread.join();
    EXPECT_TRUE(popCompleted);

    // 测试满队列的Push阻塞
    EXPECT_TRUE(queue.Push(2));
    EXPECT_TRUE(queue.Push(3));

    std::atomic<bool> pushCompleted(false);
    std::thread pushThread([&queue, &pushCompleted]() {
        EXPECT_TRUE(queue.Push(4));
        pushCompleted = true;
    });

    // 等待一小段时间，确保线程已经启动并阻塞
    std::this_thread::sleep_for(std::chrono::milliseconds(100));
    EXPECT_FALSE(pushCompleted);

    // Pop一个元素，应该让Push线程继续执行
    EXPECT_EQ(queue.Pop(), 2);

    // 等待Push线程完成
    pushThread.join();
    EXPECT_TRUE(pushCompleted);
}

int main(int argc, char **argv)
{
    ::testing::InitGoogleTest(&argc, argv);
    return RUN_ALL_TESTS();
}